package confirms;


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import contants.Constants;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class PublisherConfirms {

    private static final Integer MAX_COUNT = 100;
    /**
     * 建立连接
     *
     * @return 连接
     */
    static Connection createConnection() throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Constants.HOST);//主机ip
        factory.setPort(Constants.PORT);//端口号
        factory.setUsername(Constants.USER_NAME);//用户名
        factory.setPassword(Constants.PASSWORD);//用户密码
        factory.setVirtualHost(Constants.VIRTUAL_HOST);//虚拟主机
        return factory.newConnection();
    }



    public static void main(String[] args) throws Exception {
        //1.单独确认
        publishingMessageIndividually();
        //2.批量确认
        publishingMessageInBatches();
        //3.异步确认
        handlingPublisherConfirmsAsynchronously();
    }

    /**
     * 异步确认
     */
    private static void handlingPublisherConfirmsAsynchronously() throws Exception {
        //可以一遍发送消息一边进行确认
        try(Connection connection = createConnection()) {
            //1.开启信道
            Channel channel = connection.createChannel();
            //2.设置确认模式
            channel.confirmSelect();
            //3.指定队列
            channel.queueDeclare(Constants.PUBLISH_CONFIRMS_QUEUE3,true,false,false,null);
            //4.监听队列
            long start  = System.currentTimeMillis();

            SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());//用来存储未被确认的消息序号
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long l, boolean b) throws IOException {
                    //收到ack
                    if(b) {
                        //批量确认
                        confirmSeqNo.headSet(l+1).clear();
                    }else {
                        //单独确认
                        confirmSeqNo.remove(l);//移除id为l的消息
                    }
                }

                @Override
                public void handleNack(long l, boolean b) throws IOException {
                    //未收到ack
                    if(b) {
                        confirmSeqNo.headSet(l+1).clear();
                    }else {
                        confirmSeqNo.remove(l);
                    }
                    //
                }
            });
            //5.发送消息
            for(int i=0;i<MAX_COUNT;i++) {
                String msg = "Message #" + i;
                long seqNo = channel.getNextPublishSeqNo();//获取序号
                channel.basicPublish("",Constants.PUBLISH_CONFIRMS_QUEUE3,null,msg.getBytes());
                confirmSeqNo.add(seqNo);//存储序号
            }
            long end = System.currentTimeMillis();
            System.out.printf("异步确认策略, 消息条数: %d, 耗时: %d ms \n",MAX_COUNT, end-start);
        }
    }


    /**
     * 批量确认
     */
    private static void publishingMessageInBatches() throws Exception {
        try(Connection connection = createConnection()) {
            //1.开启信道
            Channel channel = connection.createChannel();
            //2.设置信道类型(确认模式)
            channel.confirmSelect();
            //3.声明队列
            channel.queueDeclare(Constants.PUBLISH_CONFIRMS_QUEUE2,true,false,false,null);
            //4.发送消息
            long start  = System.currentTimeMillis();
            int size = 100;//确定消息的个数
            int outstandingMessageCount = 0;//计数
            for(int i=0;i<MAX_COUNT;i++) {
                String msg = "Message #" + i;
                channel.basicPublish("",Constants.PUBLISH_CONFIRMS_QUEUE2,null,msg.getBytes());//发送消息
                outstandingMessageCount++;
                if(outstandingMessageCount == size) {
                    channel.waitForConfirmsOrDie(5000);
                    outstandingMessageCount = 0;
                }
            }
            if(outstandingMessageCount > 0) {
                channel.waitForConfirmsOrDie(5000);
            }
            long end = System.currentTimeMillis();
            System.out.printf("批量确认策略, 消息条数: %d, 耗时: %d ms \n",MAX_COUNT, end-start);
        }
    }

    /**
     * 单独确认
     */
    private static void publishingMessageIndividually() throws Exception {
            try(Connection connection = createConnection()) {
                //1.开启信道
                Channel channel = connection.createChannel();
                //2.设置信道类型(确认模式)
                channel.confirmSelect();
                //3.声明队列
                channel.queueDeclare(Constants.PUBLISH_CONFIRMS_QUEUE1,true,false,false,null);
                //4.发送消息
                long start  = System.currentTimeMillis();
                for(int i=0;i<MAX_COUNT;i++) {
                    String msg = "Message #" + i;
                    channel.basicPublish("",Constants.PUBLISH_CONFIRMS_QUEUE1,null,msg.getBytes());//发送消息
                    channel.waitForConfirmsOrDie(5000);//等待确认，超过5000ms报错
                }
                long end = System.currentTimeMillis();
                System.out.printf("单独确认策略, 消息条数: %d, 耗时: %d ms \n",MAX_COUNT, end-start);
            }
    }

}
